package com.imooc.flink.dataset;

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.types.Row;

/**
 * @author: Chen Yixing
 * @since: 2023/10/31 14:01:20
 * @description:
 */
public class SQLDataSetApp {
    private final static String TAG = "SQLDataSetApp";

    public static void main(String[] args) throws Exception {
//        testJsonFile();
        testJDBC();
    }

    private static void testJDBC() throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);
        tableEnv.executeSql("CREATE TABLE gbac (\n" +
                "  id BIGINT,\n" +
                "  name STRING,\n" +
                "  PRIMARY KEY (id) NOT ENFORCED\n" +
                ") WITH (\n" +
                "   'connector' = 'jdbc',\n" +
                "   'url' = 'jdbc:mysql://192.168.2.105:3306/gbac',\n" +
                "   'table-name' = 'ubi_user_info',\n" +
                "   'username' = 'root',\n" +
                "   'password' = 'root'\n" +
                ")");
        Table table = tableEnv.sqlQuery("select * from gbac");
        tableEnv.toRetractStream(table, Row.class).filter(x -> x.f0).print("row:");

        env.execute(TAG);
    }

    private static void testJsonFile() throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);
        tableEnv.executeSql("CREATE TABLE t1 (\n" +
                "  id BIGINT,\n" +
                "  name STRING,\n" +
                "  age STRING,\n" +
                "  PRIMARY KEY (id) NOT ENFORCED\n" +
                ") WITH (\n" +
                "  'connector' = 'filesystem',\n" +
                "  'path' = 'file:///D:/workspace/java/bigdata/hudi/data/t1.json',\n" +
                "  'format' = 'json'\n" +
                ")");
        Table table = tableEnv.sqlQuery("select * from t1");
        tableEnv.toRetractStream(table, Row.class).filter(x -> x.f0).print("row:");

        env.execute(TAG);
    }

}
